package org.seeplnframework.skyeye.common.advisor;

import com.google.common.base.Joiner;
import io.opentracing.Scope;
import io.opentracing.Span;
import io.opentracing.Tracer;
import net.bytebuddy.asm.Advice;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.seeplnframework.skyeye.common.enums.MDCConstantEnum;
import org.seeplnframework.skyeye.common.enums.RequestTypeEnum;
import org.seeplnframework.skyeye.common.util.K8sUtils;
import org.seeplnframework.skyeye.common.util.MDCUtils;
import org.seeplnframework.skyeye.common.util.SkyEyeStringValueResolverHelper;
import org.seeplnframework.skyeye.common.util.SpringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;

import java.lang.reflect.Method;

public class KafkaIMessageCallbackAdvisor {

    public static Logger log = LoggerFactory.getLogger(KafkaIMessageCallbackAdvisor.class);

    @Advice.OnMethodEnter
    public static void onMethodEnter(@Advice.This Object object,
                                     @Advice.Origin Method method,
                                     @Advice.AllArguments Object[] arguments,
                                     @Advice.Local("scope") Scope scope,
                                     @Advice.Local("span") Span span) {
        try {
            KafkaListener kl = method.getAnnotation(KafkaListener.class);
            if (kl != null) {
                //group
                if (StringUtils.isNotBlank(kl.groupId())) {
                    String groupId = SkyEyeStringValueResolverHelper.resolve(kl.groupId());
                    MDCUtils.put(MDCConstantEnum.CONSUMER_GROUP.getCode(), groupId);
                }
                //topics
                if (ArrayUtils.isNotEmpty(kl.topics())) {
                    MDCUtils.put(MDCConstantEnum.MESSAGE_TOPIC.getCode(), Joiner.on(",").join(kl.topics()));
                }
                //open tracing注入span
                Tracer tracer = SpringUtils.getBean(Tracer.class);
                if (tracer != null) {
                    Tracer.SpanBuilder spanBuilder = tracer
                            .buildSpan("kafka-consumer-custom-active-span")
                            .withTag(Joiner.on(",").join(kl.topics()), K8sUtils.getContainerName());
                    span = spanBuilder.start();
                    scope = tracer.activateSpan(span);
                }
            }
            MDCUtils.put(MDCConstantEnum.MQ_TYPE.getCode(), "kafka");
            MDCUtils.put(MDCConstantEnum.REQUEST_TYPE.getCode(), RequestTypeEnum.MQ.name().toLowerCase());
            MDCUtils.put(MDCConstantEnum.REQUEST_URI.getCode(), object.getClass().getName());
        } catch (Exception e) {
            log.warn("log skyeye >>> IMessageCallbackAdvisor.onMethodEnter occur exception", e);
        }
    }

    @Advice.OnMethodExit
    public static void onMethodExit(@Advice.Local("scope") Scope scope, @Advice.Local("span") Span span) {
        //MDCUtils.clear();

        if (scope != null) {
            scope.close();
        }
        if (span != null) {
            span.finish();
        }
    }
}
